/*
 * Copyright (c) 2007 The Trustees of Indiana University. All rights reserved.
 * IU CSCI B534 Distributed Systems 2007-02-01 http://www.cs.indiana.edu/classes/b534
 */

package edu.indiana.cs.b534.torrent.impl;

import edu.indiana.cs.b534.torrent.Constants;
import edu.indiana.cs.b534.torrent.MessageType;
import edu.indiana.cs.b534.torrent.StorageManager;
import edu.indiana.cs.b534.torrent.TorrentException;
import edu.indiana.cs.b534.torrent.Utils;
import edu.indiana.cs.b534.torrent.context.PeerDescription;
import edu.indiana.cs.b534.torrent.context.RequestBean;
import edu.indiana.cs.b534.torrent.context.TorrentContext;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.logging.Logger;

/**
 * Manages the interactions with a single peer for a particular torrent
 */
public class PeerInstance implements Runnable {

    private InputStream in = null;
    private OutputStream out = null;
    //TODO set this to maxumum size of possible message, so it will never overflow
    private byte[] buf = new byte[Constants.BLOCK_SIZE * 2];

    private final String connectingPeerId;
    private byte[] infoHash;

    private PieceInfo currentPiece;
    private final StorageManager storageManager;

    private InfoCollector infoCollector;

    private static Logger logger = Logger.getLogger(TorrentManager.TORRENT_MANAGER_NAME);

    /**
     * This is about my decisions.
     * amInterested - I expressed to the connecting peer that I am interested to it
     * amChoked - I expressed to the connecting peer that I choked it. Active voice ;)
     */
    private boolean amInterested = false, amChoked = true;

    /**
     * this is about peer's decisions
     * peerInterested - is it interested
     * peerChoked - has it choked me
     */
    private boolean peerInterested = false;
    //private boolean peerChoked = true;
    private PieceManager manager;
    private MessageUtil messageUtil;
    private int beginIndex = 0;
    private boolean busy = false;
    private final PeerDescription peerDescription;
    private final boolean isIncoming;
    private ChokingPolicyManager chokingPolicyManager;
    private TorrentContext torrentContext;
    private RequestPipeline requestPipeline;
    private String ipAddress;
    private int port;

    public PeerInstance(InputStream in, OutputStream out, PieceManager pieceManager) throws TorrentException {

        try {
            this.in = in;
            this.out = out;
            manager = pieceManager;

            continueWithConnectionRequestAndCompleteHandshake(in, pieceManager);

            this.connectingPeerId = messageUtil.getConnectingPeerId();
            torrentContext = manager.getTorrentContext();
            peerDescription = derivePeerDescription(torrentContext);
            this.storageManager = torrentContext.getStorageManager();
            this.isIncoming = true;

            chokingPolicyManager = new ChokingPolicyManager(torrentContext);
            infoCollector = torrentContext.getInfoCollector();

            if (torrentContext.getRequestPipeLineSize() > 1) {
                requestPipeline = new RequestPipeline(torrentContext.getRequestPipeLineSize(), this);
            }
        } catch (Throwable e) {
            try {
                if (in != null) {
                    in.close();
                }
                if (out != null) {
                    out.close();
                }
            } catch (IOException e1) {
            }
            throw new TorrentException(e);
        }
    }

    private PeerDescription derivePeerDescription(TorrentContext torrentContext) throws TorrentException {
        PeerDescription tempPeerDescription = torrentContext.getPeerDescription(connectingPeerId);
        if (tempPeerDescription == null) {
            tempPeerDescription = new PeerDescription(connectingPeerId);
            torrentContext.addPeerDescription(tempPeerDescription);
        }
        return tempPeerDescription;
    }

    /**
     * @param infoHash
     * @param ipAddress
     * @param port
     * @param manager
     * @param mypeerID
     * @throws TorrentException
     */
    public PeerInstance(byte[] infoHash, String ipAddress, int port, PieceManager manager, String mypeerID) throws TorrentException {
        try {
            this.infoHash = infoHash;
            this.manager = manager;
            this.isIncoming = false;

            connectAndCompleteHandshake(infoHash, mypeerID, ipAddress, port);

            this.connectingPeerId = messageUtil.getConnectingPeerId();

            torrentContext = manager.getTorrentContext();
            peerDescription = derivePeerDescription(torrentContext);
            this.storageManager = torrentContext.getStorageManager();

            chokingPolicyManager = new ChokingPolicyManager(torrentContext);
            infoCollector = torrentContext.getInfoCollector();
            if (torrentContext.getRequestPipeLineSize() > 1) {
                requestPipeline = new RequestPipeline(torrentContext.getRequestPipeLineSize(), this);
            }
            this.ipAddress = ipAddress;
            this.port =port;
        } catch (Throwable e) {
            try {
                if (in != null) {
                    in.close();
                }
                if (out != null) {
                    out.close();
                }
            } catch (IOException e1) {
            }
            throw new TorrentException(e);
        }
    }

    /**
     * This will be called when we got a request from a peer
     *
     * @param in
     * @param pieceManager
     * @throws TorrentException
     */
    private void continueWithConnectionRequestAndCompleteHandshake(InputStream in, PieceManager pieceManager) throws TorrentException {
        logger.info("Waitng for handshake message ....");
//        Utils.readGivenNumberOfBytes(in, 4, buf);
//        int msgLength = Utils.byteArrayToInt(buf, 0);
//        logger.info("Connection with a message length " + msgLength + " received from a peer");

        Utils.readGivenNumberOfBytes(in, MessageUtil.HANDSHAKE_SIZE, buf);

        messageUtil = new MessageUtil(buf, pieceManager.getTorrentContext().getInfoHash(), pieceManager.getTorrentContext().getMyPeerId());
        //TODO this does not send IP and port, what to do?
        this.infoHash = messageUtil.getInfoHash();
        logger.info("Handshake message received from " + connectingPeerId + ". Yet to send the response");

    }

    /**
     * This will be called when we are initiating the connection with another peer
     *
     * @param infoHash
     * @param mypeerID
     * @param ipAddress
     * @param port
     * @throws IOException
     * @throws TorrentException
     */
    private void connectAndCompleteHandshake(byte[] infoHash, String mypeerID, String ipAddress, int port) throws IOException, TorrentException {
        messageUtil = new MessageUtil(infoHash, mypeerID);

        //Connect to remote host
        Socket socket = new Socket(ipAddress, port);

        logger.info("Connection established with peer in " + ipAddress + ":" + port);
        in = socket.getInputStream();
        out = socket.getOutputStream();

        //send handshake to remote peer
        // handshake is first message: header, reserved bytes, infohash, peerid

        sendHandShakeMessage();

        logger.info("Handshake message sent .. ");

        // check if infohash is expected infohash
        // check if remote peerid is expected peerid
        //Read and veify the handshake
//        Utils.readGivenNumberOfBytes(in, 4, buf);
//        int messageLength = Utils.byteArrayToInt(buf, 0);
//
//        assert messageLength == messageUtil.getCapacity();


        Utils.readGivenNumberOfBytes(in, MessageUtil.HANDSHAKE_SIZE, buf);
        //TODO verify peerID and info Hash
        //throw new UnsupportedOperationException();
        messageUtil.veirfyIncomingHandshake(buf);
        logger.info("Handshake response received and verified ...");
    }


    public String getID() {
//        return new StringBuffer().append(infoHash).append(":")
//                .append(remoteIP).append(":").append(remotePort).toString();
        return this.connectingPeerId;
    }

    //TODO we need sync here if we invoke same code from the server and peer instance
    public synchronized void handleMessage(MessageType type, byte[] payload) throws TorrentException {
        int pieceIndex;
        int begin;
        int length;
        switch (type) {
            case Choke:

                logger.info("Choke message received from " + peerDescription);
                //If we got this while busy, cancel current piece 
                if (busy) {
                    manager.cancelPieceDownload(currentPiece, this);
                    busy = false;
                }

                peerDescription.setChoked(true);
                break;

            case UnChoke:
                logger.info("Un-Choke message received from " + peerDescription);

                // ask manager for a a piece to download from remote peer
                // 'request' the piece from peer and start timer
                //manager.getPieceToDownload(availableBitField)

                peerDescription.setChoked(false);
                break;

            case Interested:
                logger.info("Interested message received from " + peerDescription);

                if (manager.interestedMessageReceived(this)) {
                    amChoked = false;
                    this.sendUnChokeMessage();
                }
                peerInterested = true;

                break;

            case NotInterested:
                logger.info("Not Interested message received from " + peerDescription);

                // update interest of peer in pieces we have
                peerInterested = false;
                break;

            case Have:
                logger.info("Have message received from " + peerDescription);

                int haveIndex = Utils.byteArrayToInt(payload, 0);

                // update available bit field of pieces that remote peer has

                // TODO : It is better to put addPiece method in to the peer description, rather than
                // directly adding like this. The reason is, else things might go wrong when we get
                // duplicate have messages
                peerDescription.addPiece(haveIndex);

                break;

            case BitField:
                logger.info("BitField message received from " + peerDescription);

                // init available bit field of pieces that remote peer has
                // ask manager for a a piece to download from remote peer
                // if piece is to be downloaded, send interested message
                // if unchoked, 'request' the piece from peer and start timer
                boolean[] pieces = Utils.bitsToBooleanArray(payload);
                for (int i = 0; i < pieces.length; i++) {
                    if (pieces[i]) {
                        //TODO make sure repeated messages does not get here
                        peerDescription.addPiece(i);
                    }
                }

                break;

            case Request:
                logger.info("Request message received from " + peerDescription);

                // we calculate our choking policy for each and every request
//                if (!torrentContext.isSeed() && chokingPolicyManager.shouldIChoke(this)) {
//                    logger.info("Chocking the request from " + this.connectingPeerId + ", in accordance with our choking policy.");
//                    sendChokeMessage();
//                    amChoked = true;
//                    break;
//                }
                // get the 'piece' from manager and send it to peer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);
                length = Utils.byteArrayToInt(payload, 8);


                if (!amChoked) {

                    // For me to be not choked, then by this time the corresponding peer must have sent me
                    // interested messages and all the handshakes are done. So I don't need to explicitly
                    // check them here
                    PieceInfo pieceInfo = storageManager.getPieceInfo(pieceIndex);
                    if (pieceInfo.isCompleted()) {
//                    	if ((begin + length) <= pieceInfo.getPieceLength()) {
//
//
//                            byte[] piece = storageManager.getPiece(pieceIndex);
//                            byte[] dataToBeSent = new byte[length];
//
//                            //We do this to avoid array index out of bound on last block of the piece
//                            System.arraycopy(piece, begin, dataToBeSent, 0, length);
//                            try {
//                                this.sendMessage(messageUtil.buildPiece(pieceIndex, begin, dataToBeSent));
//                            } catch (TorrentException e) {
//                                logger.severe("Can not send piece to the peer. " + e);
//                            }
//                        } else {
//                            throw new TorrentException(connectingPeerId + " is requesting a block " +
//                                    "inside the piece " + pieceIndex + ". But I do not have a block with the beginning index " + begin + " and length " + length);
//                        }
                        // some one might ask a block which can not possibly be inside the piece. Let's check it
                        if (begin + length <= pieceInfo.getPieceLength()) {
                            RequestBean work = new RequestBean(pieceIndex, this, begin, length);
                            manager.getTorrentContext().getRequestQueue().addWork(work);
                            infoCollector.setPieceUploadedToPeer(connectingPeerId, pieceIndex);

                            logger.info("For peer " + peerDescription + " Block for piece " + pieceIndex + "from " + begin + " and length " + length + " is Added to queue");
                        } else {
                            throw new TorrentException(peerDescription + " is requesting a block " +
                                    "inside the piece " + pieceIndex + ". But I do not have a block with the beginning index " + begin + " and length " + length);
                        }

                    } else {
                        throw new TorrentException("Something really went bad, I am seding data from a piece which is not completed");
                    }
                } else {
                    logger.info("I am choked to peer " + peerDescription + ". Shutting down the connection, as I received a request message from it");
                    shutDownConnectionToPeer();
                }
                logger.info("Request processing Done");
                break;

            case Piece:

                // receive a piece from peer and send it to manager.
                // ask manager for a a piece to download from remote peer
                // 'request' the piece from peer and start timer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);
                logger.info("Piece " + pieceIndex + " received from " + peerDescription);

                if (torrentContext.getRequestPipeLineSize() > 1) {
                    requestPipeline.notifyBlockDownloaded(pieceIndex, begin, payload, 8);
                } else {
                    // save the piece
                    storageManager.storeBlock(pieceIndex, begin, payload, 8);

                    // check whether you have downloaded the whole thing, say the piece, that you are
                    // supposed to download
                    if (!storageManager.isPieceCompleted(pieceIndex)) {
                        peerDescription.setTimeLastRequestOrInterstedSent(-1);
                        // if not finish send the next request message to get the rest
                        synchronized (this) {
                            beginIndex = beginIndex + Constants.BLOCK_SIZE;
                        }
                        PieceInfo pieceInfo = storageManager.getPieceInfo(pieceIndex);
                        int amountLeftToDownload = pieceInfo.getPieceLength() - beginIndex;

                        if (amountLeftToDownload > 0) {

                            logger.info(amountLeftToDownload + "/" + pieceInfo.getPieceLength() + " left to be downloaded in piece " + pieceIndex);

                            int amountToDownLoadNext = Math.min(Constants.BLOCK_SIZE, amountLeftToDownload);
                            sendRequestMessage(pieceIndex, beginIndex, amountToDownLoadNext);
                        } else {
                            throw new TorrentException("I don't have anything to download (amount left " +
                                    "to be downloaded = " + amountLeftToDownload + "). But I was asked by " +
                                    "Storage Manager to download more data for the piece " + pieceIndex);
                        }
                    } else {
                        manager.setPieceDownloaded(manager.getPieceInfo(pieceIndex), this.connectingPeerId);
                        peerDescription.setCurrentDownloadingPiece(-1);
                        busy = false;
                    }
                    logger.info("Piece Message " + pieceIndex + " offset " + begin + " is saved");
                }
                break;
            case Cancel:
                logger.info("Cancel message received from " + peerDescription);

                // stop sending piece to peer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);
                length = Utils.byteArrayToInt(payload, 8);
                //TODO
                break;

            default:
                throw new TorrentException("Unknown message type");
        }
    }

    public synchronized void shutDownConnectionToPeer() {
        try {
            this.out.close();
            this.in.close();
            if (currentPiece != null) {
                this.manager.cancelPieceDownload(currentPiece, this);
            }
            this.manager.getTorrentContext().removePeer(this.connectingPeerId);
            this.manager.getTorrentContext().addAvailableNonConnectedPeer(this.ipAddress, this.port, this.connectingPeerId);
        } catch (IOException e) {
            logger.info("Couldn't shutdown the connection to peer " + this.connectingPeerId);
        } catch (TorrentException e) {
            logger.info("Couldn't add the peer to available list of peers " + this.connectingPeerId);

        }
    }

    public void sendHaveMessage(PieceInfo piece) throws TorrentException {
        sendMessage(messageUtil.buildHave(piece.getIndex()));
        logger.info("Have message sent ..");
        // send 'have' message for piece to remote peer
    }

    public void sendUnChokeMessage() throws TorrentException {
        sendMessage(messageUtil.buildUnChoke());
        logger.info("Unchoke message sent ..");
    }

    public void sendChokeMessage() throws TorrentException {
        sendMessage(messageUtil.buildChoke());
        logger.info("Choke message sent ..");
    }

    public void sendInterested() throws TorrentException {
        sendMessage(messageUtil.buildInterested());
        peerDescription.setSentInterested(true);
        logger.info("Interested message sent ..");
        peerDescription.setTimeLastRequestOrInterstedSent(System.currentTimeMillis());
    }

    public void run() {

        try {
            if (!isIncoming) {

                // if I have something to exchange with others, then let's send the bit field message
                boolean[] availability = storageManager.getPieceAvailability();
                if (availability != null) {
                    this.sendBitFieldMessage(availability);
                }

                // ok, now we are ready to connect with the peer. Let's send the interested message
                this.sendInterested();

            } else {
                //we have already verified the incoming handshake by this time
                sendHandShakeMessage();

                logger.info("Handshake response sent ...");

                // if I have something to exchange with others, then let's send the bit field message
                boolean[] availability = storageManager.getPieceAvailability();
                if (availability != null) {
                    this.sendBitFieldMessage(availability);
                }
            }

            // now we are done with shaking hands ;)

            //keep reading messages
            while (true) {
                byte[] buf = new byte[Constants.BLOCK_SIZE * 2];
                logger.info("Waiting for message from peer " + connectingPeerId);
                // get message length and content
                Utils.readGivenNumberOfBytes(in, 4, buf);
                int messageLength = Utils.byteArrayToInt(buf, 0);
                if (messageLength == 0) {
                    continue;
                }

                //get message type from reading first byte of message content
                // call handle message with type and rest of content as payload
                if (messageLength > 33000) {
                    System.out.println("messageLength = " + messageLength);
                }
                Utils.readGivenNumberOfBytes(in, messageLength, buf);
                byte messageType = buf[0];
                MessageType enumMessageType;
                switch (messageType) {
                    case 0:
                        enumMessageType = MessageType.Choke;
                        break;
                    case 1:
                        enumMessageType = MessageType.UnChoke;
                        break;
                    case 2:
                        enumMessageType = MessageType.Interested;
                        break;
                    case 3:
                        enumMessageType = MessageType.NotInterested;
                        break;
                    case 4:
                        enumMessageType = MessageType.Have;
                        break;
                    case 5:
                        enumMessageType = MessageType.BitField;
                        break;
                    case 6:
                        enumMessageType = MessageType.Request;
                        break;
                    case 7:
                        enumMessageType = MessageType.Piece;
                        break;
                    case 8:
                        enumMessageType = MessageType.Cancel;
                        break;
                    default:
                        throw new TorrentException("Unknown Message type " + messageType);
                }
                byte[] payload = new byte[messageLength - 1];
                System.arraycopy(buf, 1, payload, 0, messageLength - 1);
                handleMessage(enumMessageType, payload);
            }
        } catch (TorrentException e) {
            shutDownConnectionToPeer();
            e.printStackTrace();
//            try {
//				if(ipAddress != null && port != 0){
//					PeerInstance Instance = new PeerInstance(infoHash,ipAddress,port,manager,peerDescription.getPeerId());
//					manager.getTorrentContext().addPeer(Instance, false);
//				}
//			} catch (TorrentException e1) {
//				// TODO Auto-generated catch block
//				e1.printStackTrace();
//			}
        }
    }

    private void sendBitFieldMessage(boolean[] pieceAvailability) throws TorrentException {
        sendMessage(messageUtil.buildBitField(pieceAvailability));
        logger.info("Bitfiled message sent ..");
    }

    public void sendHandShakeMessage() throws TorrentException {
        try {
            byte[] content = messageUtil.buildHandShake();
            //Handshake does not have length
            out.write(content);
            out.flush();
        } catch (IOException e) {
            throw new TorrentException(e);
        }
    }


    private void sendMessage(byte[] content) throws TorrentException {
        // get content size and prepend to content
        // open socket to remoteIP:remotePort
        // send bytes to remote peer

        //TODO check does this output strream is alive or not, and reopen if it not alive
        try {
            if (content[0] >= 0 && content[0] <= 8 || content[0] == 19) {
                byte[] intAsBytes = Utils.intToByteArray(content.length);
                if(content.length != Utils.byteArrayToInt(intAsBytes, 0)){
                	System.out.println("Inter to bytes <-> is wrong");
                	System.exit(0);
                }
            	out.write(intAsBytes);
                
                
                out.write(content);
                out.flush();
            } else {
                throw new RuntimeException("Message does not starts with Message type, found  " + content[0]);
            }
        } catch (IOException e) {
            throw new TorrentException(e);
        }
    }

//    private void readAndVerifyHandshake() throws TorrentException {
//    	
//    }


    public void sendRequestMessage(int pieceIndex, int begin, int length) throws TorrentException {
        byte[] bytes = messageUtil.buildRequest(pieceIndex, begin, length);
        this.sendMessage(bytes);
        logger.info("Request message sent to download piece " + pieceIndex + " offset " + begin + " length " + length);
        peerDescription.setTimeLastRequestOrInterstedSent(System.currentTimeMillis());
    }


    public boolean isBusy() {
        return busy;
    }

    public synchronized void startDownload(int pieceIndex) throws TorrentException {
        if (busy) {
            throw new TorrentException("Trying to start new piece download when peer instance " + peerDescription + " is busy");
        } else {
            this.currentPiece = manager.getPieceInfo(pieceIndex);
            this.busy = true;

            if (torrentContext.getRequestPipeLineSize() > 1) {
                requestPipeline.startDownloadPiece(pieceIndex);
            } else {
                beginIndex = 0;
                peerDescription.setCurrentDownloadingPiece(pieceIndex);
                sendRequestMessage(pieceIndex, 0, Constants.BLOCK_SIZE);
            }
        }
    }

    public synchronized void resetBusyState() {
        this.currentPiece = null;
        this.busy = false;
        beginIndex = 0;
        peerDescription.setCurrentDownloadingPiece(-1);
    }


    public synchronized void sendPieceMessage(int pieceIndex, int begin, byte[] dataToBeSent) throws TorrentException {
        this.sendMessage(messageUtil.buildPiece(pieceIndex, begin, dataToBeSent));
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof PeerInstance) {
            return this.connectingPeerId.equals(((PeerInstance) obj).connectingPeerId);
        } else {
            return false;
        }
    }


    @Override
    public int hashCode() {
        return connectingPeerId.hashCode();
    }

    public StorageManager getStorageManager() {
        return storageManager;
    }

    public PieceManager getManager() {
        return manager;
    }

    public PeerDescription getPeerDescription() {
        return peerDescription;
    }


}
